package cn.z2huo.demo.hello.kafka.impl;

import cn.z2huo.demo.hello.core.init.InitializingTask;
import cn.z2huo.demo.hello.kafka.annotation.SubscriberListener;
import cn.z2huo.demo.hello.kafka.api.KafkaOperations;
import cn.z2huo.demo.hello.kafka.api.KafkaSubscriber;
import cn.z2huo.demo.hello.kafka.support.KafkaSubscriberRegister;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AnnotationUtils;

/**
 * <p>
 *
 * @author z2huo
 */
@Slf4j
@RequiredArgsConstructor
public class KafkaSubscriberScanner extends InitializingTask implements BeanFactoryAware {

    private DefaultListableBeanFactory beanFactory;

    private final KafkaOperations kafkaOperations;

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = (DefaultListableBeanFactory) beanFactory;
    }

    @SuppressWarnings("unchecked")
    @Override
    public void execute(ApplicationContext context) {
        context.getBeansOfType(KafkaSubscriber.class).forEach((beanName, subscriber) -> {
            SubscriberListener subscriberAnnotation = null;
            Class<? extends KafkaSubscriber> subscriberClass = null;
            try {
                subscriberClass = (Class<? extends KafkaSubscriber>) Class.forName(beanFactory.getBeanDefinition(beanName).getBeanClassName());
                subscriberAnnotation = AnnotationUtils.findAnnotation(subscriberClass, SubscriberListener.class);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                return;
            }
            if (subscriberAnnotation != null) {
                subscriber.topics().forEach(topic -> KafkaSubscriberRegister.put(topic, subscriber));
                kafkaOperations.subscribe(subscriberAnnotation.value(), subscriber);
            }
        });
    }

    @Override
    public int getOrder() {
        return Ordered.LOWEST_PRECEDENCE;
    }
}
